package com.ibm.cps.newmessage;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.ibm.cps.graphite.GraphiteClient;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.newmessage.MetadataInputSchema.FieldType;
import com.ibm.datamodel.timeseries.TimeSeriesSchema;
import com.ibm.interfaces.IMessageClient;
import com.ibm.interfaces.ITimeSeriesSchema;
import com.ibm.util.JSONUtility;
import com.ibm.util.exception.CPSException;

public class GraphiteDataSourceMetadata extends AbstractDataSourceMetadata {

	private String hostname;
	private int port;
	private String targets;
	private IMessageClient client;

	public GraphiteDataSourceMetadata(String input) throws CPSException {
		this(JSONUtility.fromObjectString(input));
	}

	public GraphiteDataSourceMetadata(ObjectNode node) throws CPSException {
		super(node);
		this.targets = JSONUtility.GetString("targets", node);
		this.hostname = JSONUtility.GetString(MessageFields.HOSTNAME, node);
		this.port = JSONUtility.GetInteger(MessageFields.PORT, node, 80);
		this.client = new GraphiteClient(hostname, port);
	}

	@Override
	public MetadataInputSchema getInputMetadataSchema() {
		String[] commonFields = { MessageFields.TYPE, MessageFields.TENANTID, MessageFields.NAME,
				MessageFields.SCHEMA_FORMAT_TYPE, "targets" };

		String[] serverFields = { MessageFields.HOSTNAME };

		MetadataInputSchema inputSchema = new MetadataInputSchema();
		inputSchema.addField(FieldType.String, commonFields);
		inputSchema.addField(FieldType.String, serverFields);
		inputSchema.addField(FieldType.Array, MessageFields.SCHEMA);
		inputSchema.addField(FieldType.Integer, MessageFields.PORT);
		return inputSchema;
	}

	@Override
	public IMessageClient getDataSourceConsumerClient() throws CPSException {
		return client;
	}

	@Override
	public ITimeSeriesSchema getTimeSeriesSchema() throws CPSException {
		JsonNode jsonSchema = JSONUtility.fromString(schemaString);
		return new TimeSeriesSchema(tenantid, schemaid, (ArrayNode) jsonSchema, -1);
	}

	@Override
	public String getSubscribeTarget() {
		return targets;
	}

}
